package org.pucc.handle

import org.apache.flink.table.api.StatementSet
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.pucc.bean.ImportConfig
import org.pucc.common.DatabaseConstant
import org.pucc.common.DatabaseConstant.SqlServerConstant

object SqlserverHandle {
  def getSqlserverDatabaseDDL(importConfig: ImportConfig): String = {
    s"CREATE DATABASE IF NOT EXISTS ${importConfig.sourceDatabase}"
  }

  def getSqlserverTableDDL(importConfig: ImportConfig): String ={
   val a =  s"""
       |    CREATE TABLE IF NOT EXISTS ${importConfig.sourceDatabase}.${importConfig.sourceTableName} (
       |       ${importConfig.sourceFields}
       |      ) WITH (
       |    'connector' = 'sqlserver-cdc',
       |    'hostname' = '${importConfig.sourceUrl}',
       |    'port' = '${importConfig.sourcePort}',
       |    'username' = '${importConfig.sourceName}',
       |    'password' = '${importConfig.sourcePw}',
       |    'database-name' = '${importConfig.sourceDatabase}',
       |    'schema-name' = '${importConfig.sourceSchema}',
       |    'table-name' = '${importConfig.sourceTableName}',
       |    'server-time-zone' = 'Asia/Shanghai'
       |)
       |""".stripMargin
    a
  }

  def getSqlserverData(tableEnv:StreamTableEnvironment,importConfig: ImportConfig) ={
    val a = s"""
               |select ${importConfig.insertFields}
               |from ${importConfig.sourceDatabase}.${importConfig.sourceTableName}
               |""".stripMargin
    /** --- 源数据 --- */
    tableEnv.executeSql(getSqlserverDatabaseDDL(importConfig))
    tableEnv.executeSql(getSqlserverTableDDL(importConfig))
    tableEnv.executeSql(a)
  }

  def startSqlserverJob(tableEnv:StreamTableEnvironment,ss: StatementSet, importConfig: ImportConfig) = {
    /** --- 源数据 --- */
    tableEnv.executeSql(getSqlserverDatabaseDDL(importConfig))
    tableEnv.executeSql(getSqlserverTableDDL(importConfig))

    /** --- Hudi --- */
    tableEnv.executeSql(HudiHandle.getHudiDatabaseDDL(importConfig))
    tableEnv.executeSql(HudiHandle.getHudiTableDDL(importConfig))

    /** --- 开启实时同步 --- */
    val a = s"""
       |insert into ${importConfig.sinkDatabase}.${importConfig.sinkTableName}
       |select ${importConfig.insertFields}
       |from ${importConfig.sourceDatabase}.${importConfig.sourceTableName}
       |""".stripMargin

    ss.addInsertSql(
      a
    )
  }

  def executeSqlserverJob(tableEnv:StreamTableEnvironment, importConfig: ImportConfig) = {
    /** --- 源数据 --- */
    tableEnv.executeSql(getSqlserverDatabaseDDL(importConfig))
    tableEnv.executeSql(getSqlserverTableDDL(importConfig))

    /** --- Hudi --- */
    tableEnv.executeSql(HudiHandle.getHudiDatabaseDDL(importConfig))
    tableEnv.executeSql(HudiHandle.getHudiTableDDL(importConfig))

    /** --- 开启实时同步 --- */
    val a = s"""
               |insert into ${importConfig.sinkDatabase}.${importConfig.sinkTableName}
               |select ${importConfig.insertFields}
               |from ${importConfig.sourceDatabase}.${importConfig.sourceTableName}
               |""".stripMargin

    tableEnv.executeSql(
      a
    )
  }

}
